"""
Case Type   : 资源池化集群按需回放实时构建
Case Name   : 主机故障->failover过程中kill新主机、备机
Create At   : 2024/05/13
Owner       : @kyleze
Description :
    1.打开按需回放实时构建
    2.主机执行tpcc业务，备机执行读业务
    3.查看备机实时构建
    4.故障主机，构建failover
    5.出现新主后，kill新主、备机，等待集群恢复
    6.恢复原主机
    7.查看集群状态与实时构建状态
Expect      :
    1.成功
    2.成功
    3.主机实时构建状态为DISABLED，BUILD_NORMAL
    4.成功
    5.集群恢复
    6.原主启动成功
    7.状态正常
History     :
    Modified by kyleze at 2025/3/10:增加多次判断，保证查询集群状态正常
"""
import os
import unittest
import time

from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.ComThread import ComThread

from yat.test import Node
from yat.test import macro


class OndemandHashmap(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info(f'----{os.path.basename(__file__)} start----')
        self.com = Common()
        self.constant = Constant()
        self.primary_sh = CommonSH('PrimaryDbUser')
        self.st1_sh = CommonSH('Standby1DbUser')
        self.st2_sh = CommonSH('Standby2DbUser')
        self.pri_root_node = Node(node="PrimaryRoot")
        self.pri_user_node = Node(node="PrimaryDbUser")
        self.st1_user_node = Node(node="Standby1DbUser")
        self.st2_user_node = Node(node="Standby2DbUser")
        self.replication_type = self.com.show_param("replication_type")
        self.recovery_parse_workers = \
            self.com.show_param("recovery_parse_workers")
        self.recovery_redo_workers = \
            self.com.show_param("recovery_redo_workers")
        self.ondemand_recovery = \
            self.com.show_param("ss_enable_ondemand_recovery")
        self.recovery_mem_size = \
            self.com.show_param("ss_ondemand_recovery_mem_size")
        self.ondemand_realtime_status = \
            self.com.show_param("ss_enable_ondemand_realtime_build")

    def test_hashmap(self):
        step = '--step1:打开按需回放实时构建 expect:成功--'
        self.log.info(step)
        ondemand_recovery_par = ["replication_type=1",
                                 "recovery_parse_workers=4",
                                 "recovery_redo_workers=4",
                                 "ss_enable_ondemand_recovery=on",
                                 "ss_ondemand_recovery_mem_size=25GB",
                                 "ss_enable_ondemand_realtime_build=on"]
        for par in ondemand_recovery_par:
            guc_res = self.primary_sh.execute_gsguc(
                "set", self.constant.GSGUC_SUCCESS_MSG, par)
            self.log.info(guc_res)
            self.assertTrue(guc_res, "执行失败" + step)
        restart_res = self.primary_sh.restart_db_cluster()
        self.log.info(restart_res)
        self.assertTrue(restart_res, "执行失败" + step)

        step = '--step2:主机执行tpcc业务，备机执行读业务 expect:成功--'
        self.log.info(step)
        tpcc_thread = ComThread(
            self.com.start_tpcc, args=(self.pri_root_node, macro.TPCC_PATH))
        tpcc_thread.setDaemon(True)
        tpcc_thread.start()
        time.sleep(20)

        step = '--step3:查看备机实时构建 ' \
               'expect:主机实时构建状态为DISABLED，BUILD_NORMAL--'
        self.log.info(step)
        primary_status = self.primary_sh.execut_db_sql(
            "select realtime_build_status from ondemand_recovery_status();")
        self.log.info(primary_status)
        self.assertIn("DISABLED", primary_status, "执行失败" + step)
        st1_status = self.st1_sh.execut_db_sql(
            "select realtime_build_status from ondemand_recovery_status();")
        self.log.info(st1_status)
        self.assertIn("BUILD_NORMAL", st1_status, "执行失败" + step)
        st2_status = self.st2_sh.execut_db_sql(
            "select realtime_build_status from ondemand_recovery_status();")
        self.log.info(st2_status)
        self.assertIn("BUILD_NORMAL", st2_status, "执行失败" + step)

        step = '--step4:故障主机，构建failover expect:成功--'
        self.log.info(step)
        mv_cmd = f"mv {macro.GAUSS_HOME}/bin/gaussdb " \
                 f"{macro.GAUSS_HOME}/bin/gaussdb_bak"
        move_res = self.pri_user_node.sh(mv_cmd).result()
        self.log.info(move_res)
        self.assertIn('', move_res, "执行失败" + step)
        kill_cmd = "ps ux | grep '\-D' | grep gaussdb| grep -v grep " \
                   "| awk '{{print $2}}' | xargs kill -9"
        kill_res = self.pri_user_node.sh(kill_cmd).result()
        self.log.info(kill_res)
        self.assertIn('', kill_res, "执行失败" + step)

        step = '--step5:出现新主后，kill新主、备机，等待集群恢复 expect:成功--'
        self.log.info(step)
        for i in range(30):
            time.sleep(2)
            res = self.primary_sh.get_dms_reformer_lock()
            self.log.info(res)
            if res:
                kill_res = self.st1_user_node.sh(kill_cmd).result()
                self.log.info(kill_res)
                kill_res = self.st2_user_node.sh(kill_cmd).result()
                self.log.info(kill_res)
                break
        for i in range(60):
            self.log.info("----第" + str(i + 1) + "次查询集群状态----")
            time.sleep(3)
            cluster_detail = self.primary_sh.get_db_cluster_status("detail")
            self.log.info(cluster_detail)
            if "Degraded" in cluster_detail and "Primary Normal" in \
                    cluster_detail:
                break
            elif i == 59:
                cluster_res = "Degraded" or "Primary Normal" in cluster_detail
                self.assertTrue(cluster_res, "执行失败" + step)

        step = "--step6:恢复原主机 expect:成功--"
        self.log.info(step)
        mv_cmd = f"mv {macro.GAUSS_HOME}/bin/gaussdb_bak " \
                 f"{macro.GAUSS_HOME}/bin/gaussdb"
        move_res = self.pri_user_node.sh(mv_cmd).result()
        self.log.info(move_res)
        self.assertIn('', move_res, "执行失败" + step)
        start_res = self.primary_sh.exec_cm_ctl('start', '-n 1')
        self.log.info(start_res)
        self.assertIn(self.constant.cm_start_node_success_msg, start_res,
                      "执行失败" + step)
        self.log.info("----等待回放状态normal----")
        for i in range(60):
            self.log.info("----第" + str(i + 1) + "次查询回放状态----")
            time.sleep(2)
            cluster_info = self.st1_sh.get_pg_controldata()
            if "normal" in cluster_info["Cluster status"]:
                break
            elif i == 59:
                self.assertIn("normal", cluster_info["Cluster status"],
                              "执行失败" + step)
        time.sleep(10)

        step = '--step7:查看集群状态与实时构建状态 expect:状态正常--'
        self.log.info(step)
        cluster_status = self.primary_sh.get_db_cluster_status('detail')
        self.log.info(cluster_status)
        self.assertTrue(cluster_status, "执行失败" + step)
        res = self.primary_sh.get_dms_reformer_lock()
        self.log.info(res)
        primary_status = self.primary_sh.execut_db_sql(
            "select realtime_build_status from ondemand_recovery_status();")
        self.log.info(primary_status)
        self.assertIn("BUILD_NORMAL", primary_status, "执行失败" + step)
        if res == '6002':
            st1_status = self.st1_sh.execut_db_sql(
                "select realtime_build_status from ondemand_recovery_status();")
            self.log.info(st1_status)
            self.assertIn("DISABLED", st1_status, "执行失败" + step)
            st2_status = self.st2_sh.execut_db_sql(
                "select realtime_build_status from ondemand_recovery_status();")
            self.log.info(st2_status)
            self.assertIn("BUILD_NORMAL", st2_status, "执行失败" + step)
        else:
            st1_status = self.st1_sh.execut_db_sql(
                "select realtime_build_status from ondemand_recovery_status();")
            self.log.info(st1_status)
            self.assertIn("BUILD_NORMAL", st1_status, "执行失败" + step)
            st2_status = self.st2_sh.execut_db_sql(
                "select realtime_build_status from ondemand_recovery_status();")
            self.log.info(st2_status)
            self.assertIn("DISABLED", st2_status, "执行失败" + step)

    def tearDown(self):
        self.log.info('----恢复环境----')
        switch_res = self.primary_sh.exec_cm_ctl('switchover', '-a')
        self.log.info(switch_res)
        ondemand_recovery_par = [
            f"replication_type={self.replication_type}",
            f"recovery_parse_workers={self.recovery_parse_workers}",
            f"recovery_redo_workers={self.recovery_redo_workers}",
            f"ss_enable_ondemand_recovery={self.ondemand_recovery}",
            f"ss_ondemand_recovery_mem_size={self.recovery_mem_size}",
            f"ss_enable_ondemand_realtime_build={self.ondemand_realtime_status}"
        ]
        for par in ondemand_recovery_par:
            guc_res = self.primary_sh.execute_gsguc(
                "set", self.constant.GSGUC_SUCCESS_MSG, par)
            self.log.info(guc_res)
            self.assertTrue(guc_res, "恢复环境失败")
        restart_res = self.primary_sh.restart_db_cluster()
        self.log.info(restart_res)
        self.assertIn(self.constant.cm_switchover_success_msg, switch_res,
                      '恢复环境失败')
        self.assertTrue(restart_res, "恢复环境失败")
        self.log.info(f'----{os.path.basename(__file__)} end')
